package com.atguigu.tingshu.search.mq;


import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class SearchReceiver {

    @Autowired
    private SearchService searchService;

    @Autowired
    private RedisTemplate<String,String> redisTemplate;

    //从生产者中获取消息,进行上架操作,上架到es中会显示在主页面
    @KafkaListener(topics = "tingshu.album.upper")
    public void upperAlbum(ConsumerRecord<String, String> record){

        Long albumId = Long.parseLong(record.value()) ;
        if (albumId != null){
            searchService.upperAlbum(albumId);
        }
    }

    //从生产者中获取消息,进行下架操作,下架会从es中删除专辑
    @KafkaListener(topics = KafkaConstant.QUEUE_ALBUM_LOWER)
    public void lowerGoods(ConsumerRecord<String, String> record){
        Long albumId = Long.parseLong(record.value());
        if (albumId != null){
            searchService.getLower(albumId);
        }
    }

    @KafkaListener(topics = KafkaConstant.QUEUE_ALBUM_STAT_UPDATE)
    public void updateAlbumStat(ConsumerRecord<String,String> record){
        String value = record.value();
        if (value != null){
            TrackStatMqVo trackStatMqVo = JSON.parseObject(value, TrackStatMqVo.class);
            String businessNo = trackStatMqVo.getBusinessNo();
            String redisKey = "album:stat:idempotent:"+businessNo;

            Boolean ifAbsent = redisTemplate.opsForValue().setIfAbsent(redisKey, businessNo, 1, TimeUnit.HOURS);
            if (ifAbsent){
                searchService.albumStatUpdate(trackStatMqVo);
            }

        }
    }
}
